package com.kitten.count.consumer;

import com.github.phantomthief.collection.BufferTrigger;
import com.google.common.collect.Lists;
import com.kitten.count.constant.MQConstants;
import com.kitten.count.constant.RedisKeyConstants;
import com.kitten.count.enums.LikeUnlikeNoteTypeEnum;
import com.kitten.count.model.dto.AggregationCountLikeUnlikeNoteMqDTO;
import com.kitten.count.model.dto.CountNoteLikeUnLikeMqDTO;
import com.kitten.framework.common.utils.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author kitten
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "kittenBook_group_" + MQConstants.TOPIC_COUNT_NOTE_LIKE,
    topic = MQConstants.TOPIC_COUNT_NOTE_LIKE
)
public class CountNoteLikeConsumer implements RocketMQListener<String> {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    private BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
            .bufferSize(50000)  // 缓存50000条
            .batchSize(1000)    // 每1000条聚合一次
            .linger(Duration.ofSeconds(5))  // 5s聚合一次
            .setConsumerEx(this::consumeMessage)
            .build();

    @Override
    public void onMessage(String body) {
        bufferTrigger.enqueue(body);
    }

    // 聚合消息
    private void consumeMessage(List<String> bodys) {
        log.info("聚合消息: {}", bodys.size());
        log.info("聚合消息内容: {}", JsonUtils.toJsonString(bodys));
        List<CountNoteLikeUnLikeMqDTO> dtos = bodys.stream().map(body -> {
            return JsonUtils.parseObject(body, CountNoteLikeUnLikeMqDTO.class);
        }).toList();
        // note点赞数据分组 18xxxxx -> [o1, o2, o3] 19xxxxx -> [o4, o5, o6]
        Map<Long, List<CountNoteLikeUnLikeMqDTO>> map = dtos.stream().collect(Collectors.groupingBy(CountNoteLikeUnLikeMqDTO::getNoteId));

        // 重构: List<AggregationCountLikeUnlikeNoteMqDTO> 存储聚合数据
        /**
         * HashMap<Long, Integer> countMap = Maps.newHashMap();    // 某note对应的点赞数count
         *         for (Map.Entry<Long, List<CountNoteLikeUnLikeMqDTO>> entry : map.entrySet()) {
         *             List<CountNoteLikeUnLikeMqDTO> list = entry.getValue();
         *             int finalCount = 0;
         *             for (CountNoteLikeUnLikeMqDTO dto : list) {
         *                 Integer type = dto.getType();
         *                 LikeUnlikeNoteTypeEnum likeUnlikeNoteTypeEnum = LikeUnlikeNoteTypeEnum.valueOf(type);
         *                 if (Objects.isNull(likeUnlikeNoteTypeEnum)) {
         *                     continue;
         *                 }
         *                 switch (likeUnlikeNoteTypeEnum) {
         *                     case LIKE -> {
         *                         finalCount++;
         *                     }
         *                     case UNLIKE -> {
         *                         finalCount--;
         *                     }
         *                 }
         *                 countMap.put(entry.getKey(), finalCount);
         *             }
         *             log.info("聚合数据: noteId: {}, 点赞数: {}", entry.getKey(), finalCount);
         *         }
         *         countMap.forEach((noteId, count) -> {
         *             String likeUnlikeRedisZSETKey = RedisKeyConstants.buildCountNoteKey(noteId);
         *             Boolean isExists = redisTemplate.hasKey(likeUnlikeRedisZSETKey);
         *             if (isExists) { // 存在Key
         *                 redisTemplate.opsForHash().increment(likeUnlikeRedisZSETKey, RedisKeyConstants.FIELD_LIKE_TOTAL, count); // key, hashKey, delta
         *                 log.info("点赞数计数 消息处理 Redis记录完成");
         *             }
         *         });
         */
        List<AggregationCountLikeUnlikeNoteMqDTO> countList = Lists.newArrayList();
        for (Map.Entry<Long, List<CountNoteLikeUnLikeMqDTO>> entry : map.entrySet()) {
            Long noteId = entry.getKey();
            List<CountNoteLikeUnLikeMqDTO> value = entry.getValue();
            int finalCount = 0;
            Long creatorId = null;
            for (CountNoteLikeUnLikeMqDTO dto : value) {
                creatorId = dto.getNoteCreatorId();

                Integer type = dto.getType();
                LikeUnlikeNoteTypeEnum likeUnlikeNoteTypeEnum = LikeUnlikeNoteTypeEnum.valueOf(type);
                if (likeUnlikeNoteTypeEnum == null) {
                    continue;
                }
                switch (likeUnlikeNoteTypeEnum) {
                    case LIKE -> {
                        finalCount++;
                    }
                    case UNLIKE -> {
                        finalCount--;
                    }
                }
            }
            countList.add(AggregationCountLikeUnlikeNoteMqDTO.builder()
                    .noteId(noteId)
                    .creatorId(creatorId)
                    .count(finalCount)
                    .build()
            );
        }
        log.info("聚合笔记点赞/取消点赞 计数数据: {}", JsonUtils.toJsonString(countList));
        // 更新RedisZSET
        countList.forEach(countItem -> {
            // 笔记发布者 ID
            Long creatorId = countItem.getCreatorId();
            // 笔记 ID
            Long noteId = countItem.getNoteId();
            // 聚合后的计数
            Integer count = countItem.getCount();

            String countNoteRedisKey = RedisKeyConstants.buildCountNoteKey(noteId);     // count:note:noteId
            Boolean isCountNoteExists = redisTemplate.hasKey(countNoteRedisKey);
            if (isCountNoteExists) {
                redisTemplate.opsForHash().increment(countNoteRedisKey, RedisKeyConstants.FIELD_LIKE_TOTAL, count); // key, hashKey, delta
            }
            // 更新 Redis 用户维度点赞数
            String countUserRedisKey = RedisKeyConstants.buildCountUserKey(creatorId);  // count:user:userId
            Boolean isCountUserExists = redisTemplate.hasKey(countUserRedisKey);
            if (isCountUserExists) {
                redisTemplate.opsForHash().increment(countUserRedisKey, RedisKeyConstants.FIELD_LIKE_TOTAL, count);
            }

        });


        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(countList)).build();
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_NOTE_LIKE_2_DB, message, new SendCallback(){
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("点赞数计数 消息处理 发送到MQ完成");
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("点赞数计数 消息处理 发送到MQ失败", throwable);
            }
        });
    }

}
